package demo02;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;

@Slf4j
public class UsersMoviesMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    private HashMap<String, String> movie_info = new HashMap<String, String>();
    private String splitter = "";
    private String movie_secondPart = "";

    public static String movieFile = "";

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        splitter = context.getConfiguration().get("SPLITTER");

       /* Path[] DistributePaths = DistributedCache.getLocalCacheFiles(context.getConfiguration());
        String line = "";
        BufferedReader br = null;
        for (Path path : DistributePaths) {
            if (path.toString().endsWith("movies.dat")) {
                br = new BufferedReader(new FileReader(path.toString()));
                while ((line = br.readLine()) != null) {
                    //movies 1::Toy Story (1995)::Animation|Children's|Comedy
                    String movieID = line.split(splitter)[0];
                    String genres = line.split(splitter)[2];
                    movie_info.put(movieID, genres);
                }
            }
        }*/

        Configuration conf = context.getConfiguration();
        log.info("配置路径：{}", conf.get("fs.defaultFS"));
        FileSystem fs = FileSystem.get(conf);
        //TODO:在实际的生产环境中，此movieFile变量无法动态获取到值
        Path filePath = new Path(StringUtils.isEmpty(movieFile) ? "/movie/movies.dat" : movieFile); // HDFS文件路径
        try (BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(filePath)))) {
            String line;
            while ((line = br.readLine()) != null) {
                //movies 1::Toy Story (1995)::Animation|Children's|Comedy
                String movieID = line.split(splitter)[0];
                String genres = line.split(splitter)[2];
                movie_info.put(movieID, genres);
            }
        }
//        fs.close();//注意：此处不能关闭
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        //users 第一个（用户ID）  1::F::1::10::48067
        //ratings 第一个（用户ID） 第二个（电影ID） 1::1193::5::978300760
        //ratings_users 1::F::1::10::48067::1193

        //movies  第一个（电影ID） 第三个（电影类型） 1::Toy Story (1995)::Animation|Children's|Comedy
        //users_movies  1::F::1::10::48067::1193::Drama
        String[] val = value.toString().split(splitter);
        movie_secondPart = movie_info.get(val[5]);
        if (movie_secondPart != null) {
            String result = value + splitter + movie_secondPart;
            log.info("合并后的数据users_movies：{}", result);
            context.write(new Text(result), NullWritable.get());
        }
    }

}
